package org.apache.rocketmq.client.java.impl;



import apache.rocketmq.v2.*;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.*;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import io.grpc.Metadata;
import io.grpc.stub.StreamObserver;
import org.apache.rocketmq.client.java.clientapis.ClientConfiguration;
import org.apache.rocketmq.client.java.clientapis.ClientException;
import org.apache.rocketmq.client.java.exception.*;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.hook.MessageInterceptor;
import org.apache.rocketmq.client.java.impl.producer.ClientSessionHandler;
import org.apache.rocketmq.client.java.message.MessageCommon;
import org.apache.rocketmq.client.java.misc.ExecutorServices;
import org.apache.rocketmq.client.java.misc.ThreadFactoryImpl;
import org.apache.rocketmq.client.java.misc.Utilities;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.route.TopicRouteData;
import org.apache.rocketmq.client.java.rpc.RpcInvocation;
import org.apache.rocketmq.client.java.rpc.Signature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkNotNull;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/10
 * @方法描述：rocketmq客户端的核心类，生产者和消费者客户端的公共功能都定义在这个类中了
 */
@SuppressWarnings({"UnstableApiUsage", "NullableProblems"})
public abstract class ClientImpl extends AbstractIdleService implements Client, ClientSessionHandler, MessageInterceptor {

    private static final Logger LOGGER = LoggerFactory.getLogger(ClientImpl.class);

    //客户端发送遥测请求的超时时间
    private static final Duration TELEMETRY_TIMEOUT = Duration.ofDays(60 * 365);

    //客户端管理器
    protected final ClientManager clientManager;

    //客户端配置信息对象
    protected final ClientConfiguration clientConfiguration;

    //客户端要访问的服务端的网络地址
    protected final Endpoints endpoints;

    //客户端订阅的主题集合
    protected final Set<String> topics;

    //这个集合存放的是可能出现了故障的服务端的网络地址
    protected final Set<Endpoints> isolated;

    //客户端回调方法的执行器
    protected final ExecutorService clientCallbackExecutor;

    //客户端处理遥测命令的执行器
    protected final ThreadPoolExecutor telemetryCommandExecutor;

    //客户端唯一Id。
    protected final String clientId;

    //客户端执行定期更新缓存的路由信息操作返回的future对象
    private volatile ScheduledFuture<?> updateRouteCacheFuture;

    //客户端缓存主题路由信息的map表
    private final ConcurrentMap<String, TopicRouteData> topicRouteCache;

    //存储正在发送查询路由信息的future对象的map表
    @GuardedBy("inflightRouteFutureLock")
    private final Map<String /* topic */, Set<SettableFuture<TopicRouteData>>> inflightRouteFutureTable;

    //保证inflightRouteFutureTable并发安全的锁
    private final Lock inflightRouteFutureLock;

    //存储客户端和不服务端建立的会话的map表
    @GuardedBy("sessionsLock")
    private final Map<Endpoints, ClientSessionImpl> sessionsTable;

    //保证sessionsTable并发安全的锁
    private final ReadWriteLock sessionsLock;

    //消息拦截器集合，在第九版本代码中并没有真的用到这个消息拦截器集合
    @GuardedBy("messageInterceptorsLock")
    private final List<MessageInterceptor> messageInterceptors;

    //保证messageInterceptors并发安全的锁
    private final ReadWriteLock messageInterceptorsLock;

    //构造方法
    public ClientImpl(ClientConfiguration clientConfiguration, Set<String> topics) {

        this.clientConfiguration = checkNotNull(clientConfiguration, "clientConfiguration should not be null");
        this.endpoints = new Endpoints(clientConfiguration.getEndpoints());
        this.topics = topics;
        //生成客户端唯一Id
        this.clientId = Utilities.genClientId();
        this.topicRouteCache = new ConcurrentHashMap<>();
        this.inflightRouteFutureTable = new ConcurrentHashMap<>();
        this.inflightRouteFutureLock = new ReentrantLock();
        this.sessionsTable = new HashMap<>();
        this.sessionsLock = new ReentrantReadWriteLock();
        this.isolated = Collections.newSetFromMap(new ConcurrentHashMap<>());
        this.messageInterceptors = new ArrayList<>();
        this.messageInterceptorsLock = new ReentrantReadWriteLock();
        this.clientManager = new ClientManagerImpl(this);

        this.clientCallbackExecutor = new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(),
                Runtime.getRuntime().availableProcessors(),
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactoryImpl("ClientCallbackWorker"));

        this.telemetryCommandExecutor = new ThreadPoolExecutor(
                1,
                1,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                new ThreadFactoryImpl("CommandExecutor"));

        //添加jvm钩子方法，确保在JVM关闭时能够正确停止客户端
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            LOGGER.info("JVM shutdown hook is invoked, clientId={}, state={}", clientId, ClientImpl.this.state());
            ClientImpl.this.stopAsync().awaitTerminated();
        }));
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：启动客户端的方法
     */
    @Override
    protected void startUp() throws Exception {

        LOGGER.info("Begin to start the rocketmq client, clientId={}", clientId);
        //在这里启动了客户端管理器，并且等待客户端管理器运行成功，然后线程才继续向下执行
        //这里的操作就会把客户端管理器中的各个定时任务提交给定时任务执行器
        this.clientManager.startAsync().awaitRunning();
        LOGGER.info("Begin to fetch topic(s) route data from remote during client startup, clientId={}, topics={}",
                clientId, topics);
        //这里执行了fetchTopicRoute方法，也就是说在客户端启动的过程中，就已经从对应的服务端获取了对应主题的路由信息
        final List<ListenableFuture<TopicRouteData>> futures =
                topics.stream().map(this::fetchTopicRoute).collect(Collectors.toList());
        //等待获取所有主题路由信息操作完成，然后当前线程才能继续向下执行
        for (ListenableFuture<TopicRouteData> future : futures) {
            future.get();
        }
        LOGGER.info("Fetch topic route data from remote successfully during startup, clientId={}, topics={}",
                clientId, topics);
        //从客户端管理器中得到定时任务执行器
        final ScheduledExecutorService scheduler = clientManager.getScheduler();
        //在这里向定时任务执行器提交定期更新缓存的主题路由信息的任务
        this.updateRouteCacheFuture = scheduler.scheduleWithFixedDelay(() -> {
            try {
                //更新缓存的主题路由信息的方法
                updateRouteCache();
            } catch (Throwable t) {
                LOGGER.error("Exception raised while updating topic route cache, clientId={}", clientId, t);
            }
        }, 10, 30, TimeUnit.SECONDS);
        LOGGER.info("The rocketmq client starts successfully, clientId={}", clientId);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：关闭客户端的方法，关闭客户端的过程中要清理释放一系列资源
     */
    @Override
    protected void shutDown() throws InterruptedException {
        LOGGER.info("Begin to shutdown the rocketmq client, clientId={}", clientId);

        //通知服务端客户端即将终止工作
        notifyClientTermination();
        //判断程序内部是否正在执行路由信息更新任务，如果更新任务存在，则取消该任务
        if (null != this.updateRouteCacheFuture) {
            //false表示如果任务正在执行，那就不会强制中断线程
            updateRouteCacheFuture.cancel(false);
        }
        //终止遥测命令执行器工作，停止接收新任务，但允许已提交的任务继续执行完成
        telemetryCommandExecutor.shutdown();
        //等待定时任务执行器中的任务完成或超时
        if (!ExecutorServices.awaitTerminated(telemetryCommandExecutor)) {
            //等待超时则记录错误日志
            LOGGER.error("[Bug] Timeout to shutdown the telemetry command executor, clientId={}", clientId);
        } else {
            //执行器正常关闭则记录成功日志
            LOGGER.info("Shutdown the telemetry command executor successfully, clientId={}", clientId);
        }
        LOGGER.info("Begin to release telemetry sessions, clientId={}", clientId);
        //释放客户端与服务端的所有会话资源
        releaseClientSessions();
        LOGGER.info("Release telemetry sessions successfully, clientId={}", clientId);
        //终止客户端管理器工作，当前线程会等待其完全终止然后才继续向下执行
        clientManager.stopAsync().awaitTerminated();
        //关闭客户端回调方法的执行器
        clientCallbackExecutor.shutdown();
        if (!ExecutorServices.awaitTerminated(clientCallbackExecutor)) {
            LOGGER.error("[Bug] Timeout to shutdown the client callback executor, clientId={}", clientId);
        }
        LOGGER.info("Shutdown the rocketmq client successfully, clientId={}", clientId);
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向客户端注册消息拦截器的方法
     */
    public void registerMessageInterceptor(MessageInterceptor messageInterceptor) {
        messageInterceptorsLock.writeLock().lock();
        try {
            messageInterceptors.add(messageInterceptor);
        } finally {
            messageInterceptorsLock.writeLock().unlock();
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该方法在第九版本代码中并没有用到，在处理消息之前，该方法就会被调用，无论是生产者发送消息，还是消费者接收消息，总是在消息被处理之前，该方法就会被调用
     * 在该方法中，每一个消息拦截器内部的方法就会被执行
     */

    @Override
    public void doBefore(MessageHookPoints hookPoint, List<MessageCommon> messageCommons) {
        messageInterceptorsLock.readLock().lock();
        try {
            for (MessageInterceptor interceptor : messageInterceptors) {
                try {
                    interceptor.doBefore(hookPoint, messageCommons);
                } catch (Throwable t) {
                    LOGGER.warn("Exception raised while intercepting message, hookPoint={}, clientId={}", hookPoint,
                            clientId);
                }
            }
        } finally {
            messageInterceptorsLock.readLock().unlock();
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：该方法在第九版本代码中并没有用到，在处理消息之后，该方法就会被调用，所谓处理消息之后，就比如说生产者客户端向服务端发送了一条消息，在发送消息之前
     * doBefore()方法会被调用，在成功发送消息接收到了服务端的响应之后，当前方法就会被调用，在该方法中，每一个消息拦截器内部的方法就会被执行
     */
    @Override
    public void doAfter(MessageHookPoints hookPoints, List<MessageCommon> messageCommons, Duration duration,
                        MessageHookPointsStatus status) {
        messageInterceptorsLock.readLock().lock();
        try {
            for (MessageInterceptor interceptor : messageInterceptors) {
                try {
                    interceptor.doAfter(hookPoints, messageCommons, duration, status);
                } catch (Throwable t) {
                    LOGGER.warn("Exception raised while intercepting message, hookPoint={}, clientId={}", hookPoints,
                            clientId);
                }
            }
        } finally {
            messageInterceptorsLock.readLock().unlock();
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：创建TelemetryCommand遥测请求的方法，这个TelemetryCommand请求会封装客户端的设置信息，客户端会把这个遥测请求发送给服务端
     * 服务端接收到该请求后会保存请求中的客户端设置信息
     */
    @Override
    public TelemetryCommand settingsCommand() {
        //获取客户端设置信息，把设置信息转换为Protobuf格式
        final Settings settings = this.getClientSettings().toProtobuf();
        //构建并返回一个包含客户端设置信息的TelemetryCommand请求
        return TelemetryCommand.newBuilder().setSettings(settings).build();
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：客户端向服务端发送遥测请求的方法
     */
    @Override
    public StreamObserver<TelemetryCommand> telemetry(Endpoints endpoints,
                                                      StreamObserver<TelemetryCommand> observer) throws ClientException {
        try {
            //生成客户端的元数据信息
            final Metadata metadata = this.sign();
            //使用客户端管理器把请求发送给服务端
            return clientManager.telemetry(endpoints, metadata, TELEMETRY_TIMEOUT, observer);
        } catch (ClientException e) {
            throw e;
        } catch (Throwable t) {
            throw new InternalErrorException(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：判断指定的服务端地址是否已被废弃的方法
     */
    @Override
    public boolean isEndpointsDeprecated(Endpoints endpoints) {
        //获取客户端内部缓存的所有可路由的服务地址信息
        final Set<Endpoints> totalRouteEndpoints = getTotalRouteEndpoints();
        //判断指定的服务端地址是否在可路由的服务地址列表中，如果不在意味着该地址已废弃
        return !totalRouteEndpoints.contains(endpoints);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：等待客户端向服务端同步设置信息的操作完成的方法
     */
    @Override
    public ListenableFuture<Void> awaitSettingSynchronized() {
        return Futures.transformAsync(this.getClientSettings().arrivedFuture,
                (clientSettings) -> Futures.immediateVoidFuture(), clientCallbackExecutor);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理来自服务端的打印线程堆栈跟踪请求的方法，该方法在第九版本代码中用不上，所以我就先不添加注释了，等后面再完善
     */
    @Override
    public void onPrintThreadStackTraceCommand(Endpoints endpoints, PrintThreadStackTraceCommand command) {
        final String nonce = command.getNonce();
        Runnable task = () -> {
            try {
                final String stackTrace = Utilities.stackTrace();
                Status status = Status.newBuilder().setCode(Code.OK).build();
                ThreadStackTrace threadStackTrace = ThreadStackTrace.newBuilder().setThreadStackTrace(stackTrace)
                        .setNonce(command.getNonce()).build();
                TelemetryCommand telemetryCommand = TelemetryCommand.newBuilder()
                        .setThreadStackTrace(threadStackTrace)
                        .setStatus(status)
                        .build();
                telemetry(endpoints, telemetryCommand);
            } catch (Throwable t) {
                LOGGER.error("Failed to send thread stack trace to remote, endpoints={}, nonce={}, clientId={}",
                        endpoints, nonce, clientId, t);
            }
        };
        try {
            telemetryCommandExecutor.submit(task);
        } catch (Throwable t) {
            LOGGER.error("[Bug] Exception raised while submitting task to print thread stack trace, endpoints={}, "
                    + "nonce={}, clientId={}", endpoints, nonce, clientId, t);
        }
    }


    //获取客户端设置信息的方法
    public abstract ClientSettings getClientSettings();


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理来自服务端的同步设置信息命令的方法
     */
    @Override
    public final void onSettingsCommand(Endpoints endpoints, Settings settings) {
        //将服务端传输过来的设置信息应用到客户端中
        this.getClientSettings().applySettingsCommand(settings);
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：同步客户端设置信息到所有服务端的方法
     */
    @Override
    public void syncSettings() {
        //得到当前客户端Protobuf格式的设置信息
        final Settings settings = getClientSettings().toProtobuf();
        //把设置信息封装到遥测请求中
        final TelemetryCommand command = TelemetryCommand.newBuilder().setSettings(settings).build();
        //得到客户端缓存的所有可以路由的服务端地址信息
        final Set<Endpoints> totalRouteEndpoints = getTotalRouteEndpoints();
        //遍历所有服务端地址
        for (Endpoints endpoints : totalRouteEndpoints) {
            try {
                //调用telemetry方法把遥测请求发送给服务端
                telemetry(endpoints, command);
            } catch (Throwable t) {
                LOGGER.error("Failed to telemeter settings, clientId={}, endpoints={}", clientId, endpoints, t);
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定服务端发送遥测请求的方法
     */
    public void telemetry(Endpoints endpoints, TelemetryCommand command) {
        try {
            //得到与指定服务端建立的会话连接
            final ClientSessionImpl clientSession = getClientSession(endpoints);
            //通过会话发送请求
            clientSession.fireWrite(command);
        } catch (Throwable t) {
            LOGGER.error("Failed to fire write telemetry command, clientId={}, endpoints={}", clientId, endpoints, t);
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：释放客户端与所有服务端建立的会话资源的方法
     */
    private void releaseClientSessions() {
        sessionsLock.readLock().lock();
        try {
            sessionsTable.values().forEach(ClientSessionImpl::release);
        } finally {
            sessionsLock.readLock().unlock();
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到客户端与指定服务端建立的会话的方法
     */
    public ClientSessionImpl getClientSession(Endpoints endpoints) throws ClientException {
        sessionsLock.readLock().lock();
        try {
            //先尝试从会话表中获取对应会话对象
            final ClientSessionImpl session = sessionsTable.get(endpoints);
            //如果得到了对应的会话对象，直接返回即可
            if (null != session) {
                return session;
            }
        } finally {
            sessionsLock.readLock().unlock();
        }
        //如果会话对象并不存在，那接下来就要获取写锁，创建对应的会话对象
        sessionsLock.writeLock().lock();
        try {
            //双重判断
            ClientSessionImpl session = sessionsTable.get(endpoints);
            if (null != session) {
                return session;
            }
            //如果会话对象仍然不存在，那就创建新的会话对象
            session = new ClientSessionImpl(this, endpoints);
            //把新创建的会话对象保存到会话表中
            sessionsTable.put(endpoints, session);
            return session;
        } finally {
            sessionsLock.writeLock().unlock();
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：安全同步客户端设置信息到指定服务端的方法
     */
    private ListenableFuture<Void> syncSettingsSafely(Endpoints endpoints) {
        try {
            final ClientSessionImpl clientSession = getClientSession(endpoints);
            return clientSession.syncSettingsSafely();
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理从服务端获取的指定主题路由信息的方法
     */
    public ListenableFuture<TopicRouteData> onTopicRouteDataFetched(String topic, TopicRouteData topicRouteData) {
        //从主题路由信息中得到最新所有可路由的服务端的网络地址
        final Set<Endpoints> routeEndpoints = topicRouteData
                .getMessageQueues().stream()
                .map(mq -> mq.getBroker().getEndpoints())
                .collect(Collectors.toSet());
        //得到当前客户端内部缓存的所有可路由的服务器的网络地址
        final Set<Endpoints> existRouteEndpoints = getTotalRouteEndpoints();
        //根据新发现的服务端地址集合和已知的服务端地址集合进行差集计算
        final Set<Endpoints> newEndpoints = new HashSet<>(Sets.difference(routeEndpoints,
                existRouteEndpoints));
        //当客户端把自己的设置信息同步给上面计算得到的newEndpoints网络地址，这里会返回一个future对象集合
        //现在我想说的是，这里的future集合中的每一个future都会在ProducerSettings对象的applySettingsCommand方法中被设置为完成状态
        //当然，这肯定是在接收到服务端响应之后了
        final List<ListenableFuture<Void>> futures =
                newEndpoints.stream().map(this::syncSettingsSafely).collect(Collectors.toList());
        //当所有同步操作完成后，这个时候就可以把客户端最新获得的路由信息缓存到本地了
        //下面whenAllSucceed方法中的回调方法就会被执行
        return Futures.whenAllSucceed(futures).callAsync(() -> {
            //在这里将获取到的最新的主题路由信息缓存起来
            topicRouteCache.put(topic, topicRouteData);
            //在这里调用子类的回调方法，可以进一步处理主题路由信息，但这个方法在第十版本代码中就会实现了
            onTopicRouteDataUpdate0(topic, topicRouteData);
            //这里再次返回了一个future对象，这个对象中封装着客户端获得的最新的主题路由信息数据
            return Futures.immediateFuture(topicRouteData);
        }, clientCallbackExecutor);
    }


    public void onTopicRouteDataUpdate0(String topic, TopicRouteData topicRouteData) {
        //该方法在第十一版本代码中就会实现
    }


    //以下两个方法在第九版本代码中并没有用到，我就不添加注释了
    @Override
    public void onVerifyMessageCommand(Endpoints endpoints, VerifyMessageCommand command) {
        LOGGER.warn("Ignore verify message command from remote, which is not expected, clientId={}, command={}",
                clientId, command);
        final String nonce = command.getNonce();
        final Status status = Status.newBuilder().setCode(Code.NOT_IMPLEMENTED).build();
        VerifyMessageResult verifyMessageResult = VerifyMessageResult.newBuilder().setNonce(nonce).build();
        TelemetryCommand telemetryCommand = TelemetryCommand.newBuilder()
                .setVerifyMessageResult(verifyMessageResult)
                .setStatus(status)
                .build();
        try {
            telemetry(endpoints, telemetryCommand);
        } catch (Throwable t) {
            LOGGER.warn("Failed to send message verification result, clientId={}", clientId, t);
        }
    }


    @Override
    public void onRecoverOrphanedTransactionCommand(Endpoints endpoints, RecoverOrphanedTransactionCommand command) {
        LOGGER.warn("Ignore orphaned transaction recovery command from remote, which is not expected, clientId={}, "
                + "command={}", clientId, command);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：客户端更新本地路由信息的方法
     */
    private void updateRouteCache() {
        LOGGER.info("Start to update route cache for a new round, clientId={}", clientId);
        //遍历客户端缓存的所有主题信息
        topicRouteCache.keySet().forEach(topic -> {
            //在这里获取每一个主题对应的最新路由信息
            final ListenableFuture<TopicRouteData> future = fetchTopicRoute(topic);
            //这里为future添加了一个回调方法，可以实现一些逻辑
            Futures.addCallback(future, new FutureCallback<TopicRouteData>() {
                @Override
                public void onSuccess(TopicRouteData topicRouteData) {
                }

                @Override
                public void onFailure(Throwable t) {
                    LOGGER.error("Failed to fetch topic route for update cache, topic={}, clientId={}", topic,
                            clientId, t);
                }
            }, MoreExecutors.directExecutor());
        });
    }


    //包装客户端终止工作请求的抽象方法
    public abstract NotifyClientTerminationRequest wrapNotifyClientTerminationRequest();


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：通知服务端客户端即将终止工作的方法
     */
    private void notifyClientTermination() {
        LOGGER.info("Notify remote that client is terminated, clientId={}", clientId);
        //得到所有可路由的服务端地址
        final Set<Endpoints> routeEndpointsSet = getTotalRouteEndpoints();
        //包装客户端终止工作请求
        final NotifyClientTerminationRequest notifyClientTerminationRequest = wrapNotifyClientTerminationRequest();
        try {
            //生成请求签名对象，用于客户端身份验证
            final Metadata metadata = sign();
            //遍历所有服务端地址
            for (Endpoints endpoints : routeEndpointsSet) {
                //向每个服务端发送客户端终止工作请求
                clientManager.notifyClientTermination(endpoints, metadata, notifyClientTerminationRequest, clientConfiguration.getRequestTimeout());
            }
        } catch (Throwable t) {
            LOGGER.error("Exception raised while notifying client's termination, clientId={}", clientId, t);
        }
    }

    //获取客户端唯一Id
    @Override
    public String clientId() {
        return clientId;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：执行客户端心跳操作的方法
     */
    @Override
    public void doHeartbeat() {
        final Set<Endpoints> totalEndpoints = getTotalRouteEndpoints();
        final HeartbeatRequest request = wrapHeartbeatRequest();
        for (Endpoints endpoints : totalEndpoints) {
            //向指定的服务端发送心跳请求
            doHeartbeat(request, endpoints);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：获取客户端签名的方法
     */
    protected Metadata sign() throws NoSuchAlgorithmException, InvalidKeyException {
        return Signature.sign(clientConfiguration, clientId);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：向指定的服务端发送心跳请求的方法
     */
    private void doHeartbeat(HeartbeatRequest request, final Endpoints endpoints) {
        try {
            //在这里得到客户端签名对象
            Metadata metadata = sign();
            //使用客户端管理器异步发送心跳请求给指定服务端，这里返回了一个future对象
            final ListenableFuture<RpcInvocation<HeartbeatResponse>> future = clientManager
                    .heartbeat(endpoints, metadata, request, clientConfiguration.getRequestTimeout());
            //向上面返回的future对象中设置回调方法
            Futures.addCallback(future, new FutureCallback<RpcInvocation<HeartbeatResponse>>() {
                //心跳请求完成之后，也就是收到了服务端的响应，上面的future会被设置为完成状态
                //这个时候回调方法就会被回调
                @Override
                public void onSuccess(RpcInvocation<HeartbeatResponse> inv) {
                    //心跳请求成功了则从RpcInvocation中获得心跳响应对象
                    //这里我多说一句服务端回复给客户端的响应会被封装到RpcInvocation对象中，这一点在RpcClientImpl类中就会得到印证
                    final HeartbeatResponse response = inv.getResponse();
                    //得到响应状态
                    final Status status = response.getStatus();
                    //得到状态码
                    final Code code = status.getCode();
                    //判断响应状态码是否为OK
                    if (Code.OK != code) {
                        //如果响应状态码不正常，则记录告警日志，意味着心跳操作失败
                        LOGGER.warn("Failed to send heartbeat, code={}, status message=[{}], endpoints={}, clientId={}",
                                code, status.getMessage(), endpoints, clientId);
                        return;
                    }
                    //如果响应状态码正常，则记录成功发送心跳的日志
                    LOGGER.info("Send heartbeat successfully, endpoints={}, clientId={}", endpoints, clientId);
                    //心跳成功则意味着和服务端建立连接成功，如果当前访问的服务端处于隔离集合中，则从隔离集合中移除即可
                    //表示当前访问的服务端可以正常访问了
                    final boolean removed = isolated.remove(endpoints);
                    if (removed) {
                        LOGGER.info("Rejoin endpoints which is isolated before, clientId={}, endpoints={}", clientId, endpoints);
                    }
                }

                //心跳请求失败时执行该回调方法
                @Override
                public void onFailure(Throwable t) {
                    LOGGER.warn("Failed to send heartbeat, endpoints={}, clientId={}", endpoints, clientId, t);
                }
            }, MoreExecutors.directExecutor());
        } catch (Throwable e) {
            LOGGER.error("Exception raised while preparing heartbeat, endpoints={}, clientId={}", endpoints, clientId,
                    e);
        }
    }

    //包装心跳请求的方法
    public abstract HeartbeatRequest wrapHeartbeatRequest();


    //执行客户端统计操作的方法，该方法在后面版本代码中才会实现
    @Override
    public void doStats() {

    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从服务端获取指定主题的路由信息的方法
     */
    private ListenableFuture<TopicRouteData> fetchTopicRoute(final String topic) {
        //执行fetchTopicRoute0()方法从服务端获取指定主题路由信息，该方法返回一个ListenableFuture对象
        //后面定义的这个topicRouteData -> onTopicRouteDataFetched(topic, topicRouteData), MoreExecutors.directExecutor()回调方法
        //会在客户端收到服务端返回的主题路由信息后执行，onTopicRouteDataFetched()方法会处理从客户端接收到的主题路由信息
        //然后把路由信息缓存在自己内部
        final ListenableFuture<TopicRouteData> future = Futures.transformAsync(fetchTopicRoute0(topic),
                topicRouteData -> onTopicRouteDataFetched(topic, topicRouteData), MoreExecutors.directExecutor());
        //为future设置回调方法
        Futures.addCallback(future, new FutureCallback<TopicRouteData>() {

            @Override
            public void onSuccess(TopicRouteData topicRouteData) {
                //获取路由信息操作成功后执行的回调方法
                LOGGER.info("Fetch topic route successfully, clientId={}, topic={}, topicRouteData={}", clientId,
                        topic, topicRouteData);
            }
            //获取路由信息操作失败后执行的回调方法
            @Override
            public void onFailure(Throwable t) {
                LOGGER.error("Failed to fetch topic route, clientId={}, topic={}", clientId, topic, t);
            }
        }, MoreExecutors.directExecutor());
        return future;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：从服务端获取指定主题路由信息的方法
     */
    private ListenableFuture<TopicRouteData> fetchTopicRoute0(final String topic) {
        try {
            //创建主题资源对象，把主题信息封装到资源对象中
            Resource topicResource = Resource.newBuilder().setName(topic).build();
            //创建查询主题路由信息请求
            final QueryRouteRequest request = QueryRouteRequest.newBuilder().setTopic(topicResource)
                    .setEndpoints(endpoints.toProtobuf()).build();
            //生成客户端签名对象
            final Metadata metadata = sign();
            //在这里使用客户端管理器把请求发送给服务端
            final ListenableFuture<RpcInvocation<QueryRouteResponse>> future =
                    clientManager.queryRoute(endpoints, metadata, request, clientConfiguration.getRequestTimeout());
            //当服务端给客户端回复了响应之后，也就是上面操作返回的future被设置为完成状态之后，下面定义的回调方法就会被执行了
            return Futures.transformAsync(future, invocation -> {
                //得到服务端回复的响应
                final QueryRouteResponse response = invocation.getResponse();
                //得到请求Id
                final String requestId = invocation.getContext().getRequestId();
                //得到响应状态
                final Status status = response.getStatus();
                //得到状态描述信息
                final String statusMessage = status.getMessage();
                //得到状态码
                final Code code = status.getCode();
                //得到状态码的值
                final int codeNumber = code.getNumber();
                //根据状态码的值执行不同操作
                switch (code) {
                    case OK:
                        //如果状态码正常，就什么也不做
                        break;
                    case BAD_REQUEST:
                    case ILLEGAL_ACCESS_POINT:
                    case ILLEGAL_TOPIC:
                    case CLIENT_ID_REQUIRED:
                        //如果状态码为请求错误，或者参数错误、非法访问、非法主题、缺少客户端Id等等
                        //则在这里抛出对应的异常
                        throw new BadRequestException(codeNumber, requestId, statusMessage);
                    case NOT_FOUND:
                    case TOPIC_NOT_FOUND:
                        //如果状态码为为资源不存在、主题不存在，抛出NotFoundException异常
                        throw new NotFoundException(codeNumber, requestId, statusMessage);
                    case TOO_MANY_REQUESTS:
                        //执行到这里意味着状态码是请求过多，服务端限流
                        throw new TooManyRequestsException(codeNumber, requestId, statusMessage);
                    case INTERNAL_ERROR:
                    case INTERNAL_SERVER_ERROR:
                        //执行到这里意味着状态码为服务端内部错误
                        throw new InternalErrorException(codeNumber, requestId, statusMessage);
                    case PROXY_TIMEOUT:
                        //执行到这里意味着状态码为代理超时
                        throw new ProxyTimeoutException(codeNumber, requestId, statusMessage);
                    default:
                        //意义未明确的状态码
                        throw new UnsupportedException(codeNumber, requestId, statusMessage);
                }
                //从响应中获取服务端返回的消息队列列表
                final List<MessageQueue> messageQueuesList = response.getMessageQueuesList();
                //创建一个包含消息队列列表的TopicRouteData对象
                final TopicRouteData topicRouteData = new TopicRouteData(messageQueuesList);
                //返回包含主题路由数据的，并且状态为已完成的future对象
                return Futures.immediateFuture(topicRouteData);
            }, MoreExecutors.directExecutor());
        } catch (Throwable t) {
            return Futures.immediateFailedFuture(t);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：获取客户端内部缓存的所有可路由的服务地址信息的方法
     */
    protected Set<Endpoints> getTotalRouteEndpoints() {
        Set<Endpoints> totalRouteEndpoints = new HashSet<>();
        for (TopicRouteData topicRouteData : topicRouteCache.values()) {
            totalRouteEndpoints.addAll(topicRouteData.getTotalEndpoints());
        }
        return totalRouteEndpoints;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：获取指定主题路由信息的方法，该方法在第九版本代码中还用不到，我就先不添加注释了，在第十版本代码中就会为大家实现了
     */
    protected ListenableFuture<TopicRouteData> getRouteData(final String topic) {
        SettableFuture<TopicRouteData> future0 = SettableFuture.create();
        TopicRouteData topicRouteData = topicRouteCache.get(topic);
        if (null != topicRouteData) {
            future0.set(topicRouteData);
            return future0;
        }
        inflightRouteFutureLock.lock();
        try {
            topicRouteData = topicRouteCache.get(topic);
            if (null != topicRouteData) {
                future0.set(topicRouteData);
                return future0;
            }
            Set<SettableFuture<TopicRouteData>> inflightFutures = inflightRouteFutureTable.get(topic);
            if (null != inflightFutures) {
                inflightFutures.add(future0);
                return future0;
            }
            inflightFutures = new HashSet<>();
            inflightFutures.add(future0);
            inflightRouteFutureTable.put(topic, inflightFutures);
        } finally {
            inflightRouteFutureLock.unlock();
        }
        final ListenableFuture<TopicRouteData> future = fetchTopicRoute(topic);
        Futures.addCallback(future, new FutureCallback<TopicRouteData>() {

            @Override
            public void onSuccess(TopicRouteData topicRouteData) {
                inflightRouteFutureLock.lock();
                try {
                    final Set<SettableFuture<TopicRouteData>> newFutureSet =
                            inflightRouteFutureTable.remove(topic);
                    if (null == newFutureSet) {
                        LOGGER.error("[Bug] in-flight route futures was empty, topic={}, clientId={}", topic,
                                clientId);
                        return;
                    }
                    LOGGER.debug("Fetch topic route successfully, topic={}, in-flight route future "
                            + "size={}, clientId={}", topic, newFutureSet.size(), clientId);
                    for (SettableFuture<TopicRouteData> newFuture : newFutureSet) {
                        newFuture.set(topicRouteData);
                    }
                } catch (Throwable t) {
                    LOGGER.error("[Bug] Exception raised while update route data, topic={}, clientId={}", topic,
                            clientId, t);
                } finally {
                    inflightRouteFutureLock.unlock();
                }
            }


            @Override
            public void onFailure(Throwable t) {
                inflightRouteFutureLock.lock();
                try {
                    final Set<SettableFuture<TopicRouteData>> newFutureSet =
                            inflightRouteFutureTable.remove(topic);
                    if (null == newFutureSet) {
                        LOGGER.error("[Bug] in-flight route futures was empty, topic={}, clientId={}", topic, clientId);
                        return;
                    }
                    LOGGER.debug("Failed to fetch topic route, topic={}, in-flight route future " +
                            "size={}, clientId={}", topic, newFutureSet.size(), clientId, t);
                    for (SettableFuture<TopicRouteData> future : newFutureSet) {
                        future.setException(t);
                    }
                } finally {
                    inflightRouteFutureLock.unlock();
                }
            }
        }, MoreExecutors.directExecutor());
        return future0;
    }


    //获取定时任务执行器的方法
    public ScheduledExecutorService getScheduler() {
        return clientManager.getScheduler();
    }

    //该方法也是在第十版本代码中才会使用，这里我就先不添加注释了
    protected <T> T handleClientFuture(ListenableFuture<T> future) throws ClientException {
        try {
            return future.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            final Throwable cause = e.getCause();
            if (cause instanceof ClientException) {
                throw (ClientException) cause;
            }
            if (cause instanceof RuntimeException) {
                throw (RuntimeException) cause;
            }
            throw new ClientException(null == cause ? e : cause);
        }
    }

    //获取客户端配置信息对象的方法
    public ClientConfiguration getClientConfiguration() {
        return clientConfiguration;
    }
}

